package test.top1.com.atguigu.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import test.top1.com.atguigu.utils.MyKafkaUtil;
import test.top1.com.atguigu.utils.MySqlUtil;

/**
 * ClassName: Dwd02_TradeCartAdd
 * Package: test.top1.com.atguigu.app.dwd.db
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/4 9:30
 * @Version 1.0
 */
public class Dwd02_TradeCartAdd {
    public static void main(String[] args) {

        //TODO 1.创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        // TODO 配置环境信息生产环境得有
//         env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 2.从卡夫卡读取topic_db 数据
       tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_cart_add_221109"));

        //TODO 3.过滤出加购数据
        Table cartInfoTable = tableEnv.sqlQuery("" +
                "select \n" +
                "    `data`[`id`] id,\n" +
                "    `data`[`user_id`] user_id,\n" +
                "    `data`[`sku_id`] sku_id,\n" +
                "    `data`[`cart_price`] cart_price,\n" +
                "    `data`[`sku_num`] sku_num,\n" +
                "    `data`[`img_url`] img_url,\n" +
                "    `data`[`sku_name`] sku_name,\n" +
                "    `data`[`is_checked`] is_checked,\n" +
                "    `data`[`create_time`] create_time,\n" +
                "    `data`[`operate_time`] operate_time,\n" +
                "    `data`[`is_ordered`] is_ordered,\n" +
                "    `data`[`order_time`] order_time,\n" +
                "    `data`[`source_type`] source_type,\n" +
                "    `data`['source_id'] source_id,\n" +
                "    `pt`\n" +
                "from topic_db\n" +
                "where `database`=`gmall`\n" +
                "and `table`=`cart_info`\n" +
                "and `type`=`insert`");
        //创建临时视图表
        tableEnv.createTemporaryView("cart_info",cartInfoTable);

        //TODO 4.从mysql读取lookup表
        tableEnv.executeSql(MySqlUtil.getMysqlBaseDicDDL());

        //TODO 5.关联两表
        Table resultTable = tableEnv.sqlQuery("" +
                "select \n" +
                "    cart_info.id,\n" +
                "    cart_info.user_id,\n" +
                "    cart_info.sku_id,\n" +
                "    cart_info.cart_price,\n" +
                "    cart_info.sku_num,\n" +
                "    cart_info.img_url,\n" +
                "    cart_info.sku_name,\n" +
                "    cart_info.is_checked,\n" +
                "    cart_info.create_time,\n" +
                "    cart_info.operate_time,\n" +
                "    cart_info.is_ordered,\n" +
                "    cart_info.order_time,\n" +
                "    cart_info.source_type,\n" +
                "    cart_info.source_id,\n" +
                "    dic.dic_name\n" +
                "from cart_info \n" +
                "join base_dic FOR SYSTEM_TIME AS OF cart_info.pt AS dic\n" +
                "on cart_info.`source_type`=dic.`dic_code`");
        tableEnv.createTemporaryView("result_table", resultTable);

        //TODO 6.6.创建kafka dwd层加购事务主题
        tableEnv.executeSql("" +
                "create table dwd_cart_info(\n" +
                "id,\n" +
                "user_id,\n" +
                "sku_id,\n" +
                "cart_price,\n" +
                "sku_num,\n" +
                "img_url,\n" +
                "sku_name,\n" +
                "is_checked,\n" +
                "create_time,\n" +
                "operate_time,\n" +
                "is_ordered,\n" +
                "order_time,\n" +
                "source_type,\n" +
                "source_id,\n" +
                "dic_name\n" +
                ")" + MyKafkaUtil.getKafkaSinkConnOption("dwd_trade_cart_add"));

        //TODO 7.插入数据
        tableEnv.executeSql("insert into dwd_cart_info select * from result_table");
    }
}
